package test.top2.com.atguigu.app.fun;

import com.alibaba.druid.pool.DruidPooledConnection;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import test.top2.com.atguigu.bean.TableProcess;
import test.top2.com.atguigu.common.GmallConfig;
import test.top2.com.atguigu.utils.DruidDSUtil;
import test.top2.com.atguigu.utils.JdbcUtil;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.*;

/**
 * ClassName: DimDwdTableProcessFunction
 * Package: test.top2.com.atguigu.app.fun
 * Description:
 *
 * @Author ChenJun(有志男青年)
 * @Create 2023/5/11 9:40
 * @Version 1.0
 */
public class DimDwdTableProcessFunction  extends BroadcastProcessFunction<JSONObject, String, JSONObject> {

    private HashMap<String, TableProcess> tableProcessHashMap;
    private MapStateDescriptor<String, TableProcess> mapStateDescriptor;

    private OutputTag<JSONObject> outputTag;
    public DimDwdTableProcessFunction(MapStateDescriptor<String, TableProcess> mapStateDescriptor, OutputTag<JSONObject> outputTag) {
        this.mapStateDescriptor = mapStateDescriptor;
        this.outputTag = outputTag;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        tableProcessHashMap = new HashMap<>();

        Connection connection = DriverManager.getConnection("jdbc:mysql://hadoop102:3306/gmall-config?" +
                "user=root&password=000000&useUnicode=true&" +
                "characterEncoding=utf8&serverTimeZone=Asia/Shanghai&useSSL=false");

        List<TableProcess> tableProcesses = JdbcUtil.queryList(connection,
                "select * from table_process",
                TableProcess.class,
                true);

        for (TableProcess tableProcess : tableProcesses) {

            String sinkType = tableProcess.getSinkType();

            if ("dim".equals(sinkType)) {
                //建表并将其放入Map
                checkTable(tableProcess);
                String key = tableProcess.getSourceTable();
                tableProcessHashMap.put(key, tableProcess);
            } else {
                //将其放入Map
                String key = tableProcess.getSourceTable() + "-" + tableProcess.getSourceType();
                tableProcessHashMap.put(key, tableProcess);
            }
        }
        connection.close();
    }



    //value:{"before":{"source_table":"base_category3","sink_table":"dim_base_category3","sink_columns":"id,name,category2_id","sink_pk":"id","sink_extend":null},"after":{"source_table":"base_category3","sink_table":"dim_base_category3","sink_columns":"id,name,category2_id","sink_pk":"id","sink_extend":null},"source":{"version":"1.5.4.Final","connector":"mysql","name":"mysql_binlog_source","ts_ms":1669162876406,"snapshot":"false","db":"gmall-220623-config","sequence":null,"table":"table_process","server_id":0,"gtid":null,"file":"","pos":0,"row":0,"thread":null,"query":null},"op":"r","ts_ms":1669162876406,"transaction":null}
    @Override
    public void processBroadcastElement(String value, BroadcastProcessFunction<JSONObject, String, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
//获取广播状态
        BroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(mapStateDescriptor);

        //如果为删除操作,则删除广播状态以及HashMap中的数据
        JSONObject jsonObject = JSON.parseObject(value);
        String op = jsonObject.getString("op");
        if ("d".equals(op)) {
            TableProcess tableProcess = JSON.parseObject(jsonObject.getString("before"), TableProcess.class);

            String sourceTable = tableProcess.getSourceTable();
            String sourceType = tableProcess.getSourceType();
            String key = sourceTable + "-" + sourceType;

            if ("dim".equals(tableProcess.getSinkType())) {
                broadcastState.remove(sourceTable);
                tableProcessHashMap.remove(sourceTable);
            } else {
                broadcastState.remove(key);
                tableProcessHashMap.remove(key);
            }
        } else {
            TableProcess tableProcess = JSON.parseObject(jsonObject.getString("after"), TableProcess.class);
            String sinkType = tableProcess.getSinkType();
            if ("dim".equals(sinkType)) {
                //建表并将其放入Map
                checkTable(tableProcess);
                String key = tableProcess.getSourceTable();
                broadcastState.put(key, tableProcess);
            } else {
                //将其放入Map
                String key = tableProcess.getSourceTable() + "-" + tableProcess.getSourceType();
                broadcastState.put(key, tableProcess);
            }
        }
    }


    //Value:{"database":"gmall-221109-flink","table":"comment_info","type":"insert","ts":1669162958,"xid":1111,"xoffset":13941,"data":{"id":1595211185799847960,"user_id":119,"nick_name":null,"head_img":null,"sku_id":31,"spu_id":10,"order_id":987,"appraise":"1204","comment_txt":"评论内容：48384811984748167197482849234338563286217912223261","create_time":"2022-08-02 08:22:38","operate_time":null}}
    @Override
    public void processElement(JSONObject value, BroadcastProcessFunction<JSONObject, String, JSONObject>.ReadOnlyContext ctx, Collector<JSONObject> out) throws Exception {
        //获取DIM以及DWD对应的Key值
        String dimKey = value.getString("table");
        String type = value.getString("type");
        String dwdKey = dimKey + "-" + type;

        //尝试获取配置信息
        ReadOnlyBroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(mapStateDescriptor);

        TableProcess dimTableProcess = broadcastState.get(dimKey);
        TableProcess dimTableProcessMap = tableProcessHashMap.get(dimKey);

        if ((dimTableProcess != null || dimTableProcessMap != null) && ("insert".equals(type) || "update".equals(type) || "bootstrap-insert".equals(type))) {
            if (dimTableProcess == null) {
                dimTableProcess = dimTableProcessMap;
            }
            //2.列过滤
            filterColumns(value.getJSONObject("data"), dimTableProcess.getSinkColumns());
            //3.写出
            value.put("sink_table", dimTableProcess.getSinkTable());
            ctx.output(outputTag, value);
        } else {
            System.out.println("该维度表:" + dimKey + "没有对应的配置信息！或者类型不匹配:" + type);
        }

        TableProcess dwdTableProcess = broadcastState.get(dwdKey);
        TableProcess dwdTableProcessMap = tableProcessHashMap.get(dwdKey);
        //2.1 过滤数据(行过滤)
        if (dwdTableProcess != null || dwdTableProcessMap != null) {
            if (dwdTableProcess == null) {
                dwdTableProcess = dwdTableProcessMap;
            }
            //2.2 过滤数据(列过滤)
            filterColumns(value.getJSONObject("data"), dwdTableProcess.getSinkColumns());
            //3.补充SinkTable字段将数据写出
            value.put("sink_table", dwdTableProcess.getSinkTable());
            out.collect(value);
        } else {
            System.out.println("该事实表：" + dwdKey + "没有对应的配置信息！");
        }
    }

    private void checkTable(TableProcess tableProcess) {

        //获取连接
        DruidPooledConnection phoenixConn = null;
        PreparedStatement preparedStatement = null;

        //获取建表相关字段
        String sinkTable = tableProcess.getSinkTable();
        String sinkPk = tableProcess.getSinkPk();
        String sinkColumns = tableProcess.getSinkColumns();
        String sinkExtend = tableProcess.getSinkExtend();

        try {
            phoenixConn = DruidDSUtil.getPhoenixConn();
            //封装建表语句 create table if not exists db.sinkTable(id varchar primary key,name varchar,sex varchar) xxx

            if (sinkPk == null || "".equals(sinkPk)) {
                sinkPk = "id";
            }
            if (sinkExtend == null) {
                sinkExtend = "";
            }
            StringBuilder createTableSql = new StringBuilder("create table if not exists ")
                    .append(GmallConfig.HBASE_SCHEMA)
                    .append(".")
                    .append(sinkTable)
                    .append("(");
            String[] columnArr = sinkColumns.split(",");
            for (int i = 0; i < columnArr.length; i++) {
                String column = columnArr[i];
                if (sinkPk.equals(column)) {
                    createTableSql.append(column).append(" varchar primary key");
                } else {
                    createTableSql.append(column).append(" varchar");
                }

                if (i < columnArr.length - 1) {
                    createTableSql.append(",");
                } else {
                    createTableSql.append(")").append(sinkExtend);
                }
            }

            //打印SQL语句
            System.out.println("建表语句为：" + createTableSql);

            //预编译SQL
            preparedStatement = phoenixConn.prepareStatement(createTableSql.toString());

            //执行建表操作
            preparedStatement.execute();

        } catch (SQLException e) {
            e.printStackTrace();
            throw new RuntimeException("建表" + sinkTable + "失败!");
        } finally {
            //归还连接
            if (preparedStatement != null) {
                try {
                    preparedStatement.close();
                } catch (SQLException e) {
                    e.printStackTrace();
                }
            }

            if (phoenixConn != null) {
                try {
                    phoenixConn.close();
                } catch (SQLException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    /**
     * 列过滤
     *
     * @param data        {"id":"1001","name":"zs","sex":"male","tm_name":"aaa"}
     * @param sinkColumns id,tm_name
     */
    private void filterColumns(JSONObject data, String sinkColumns) {

        //将sinkColumns切分并转换为集合
        String[] columns = sinkColumns.split(",");
        List<String> columnList = Arrays.asList(columns);

//        Set<Map.Entry<String, Object>> entries = data.entrySet();
//        Iterator<Map.Entry<String, Object>> iterator = entries.iterator();
//        while (iterator.hasNext()) {
//            Map.Entry<String, Object> next = iterator.next();
//            if (!columnList.contains(next.getKey())) {
//                iterator.remove();
//            }
//        }

        Set<Map.Entry<String, Object>> entries = data.entrySet();
        entries.removeIf(next -> !columnList.contains(next.getKey()));
    }


}
